package com.flink.hbase.kafka2elasticsearch.timepartition;

import java.io.Serializable;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.YearMonth;
import java.time.format.DateTimeFormatter;
import java.util.Objects;

/**
 * 时间分区信息管理
 * 用于管理基于时间的ES索引分区
 */
public class TimePartitionInfo implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private final YearMonth yearMonth;
    private final String baseIndexName;
    private final String aliasName;
    private final String actualIndexName;
    
    private static final DateTimeFormatter MONTH_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM");
    private static final DateTimeFormatter INDEX_FORMATTER = DateTimeFormatter.ofPattern("yyyy_MM");
    
    public TimePartitionInfo(String baseIndexName, YearMonth yearMonth) {
        this.baseIndexName = baseIndexName;
        this.yearMonth = yearMonth;
        this.aliasName = baseIndexName;
        this.actualIndexName = baseIndexName + "_" + yearMonth.format(INDEX_FORMATTER);
    }
    
    public TimePartitionInfo(String baseIndexName, LocalDateTime businessTime) {
        this(baseIndexName, YearMonth.from(businessTime));
    }
    
    public TimePartitionInfo(String baseIndexName, LocalDate businessDate) {
        this(baseIndexName, YearMonth.from(businessDate));
    }
    
    public YearMonth getYearMonth() {
        return yearMonth;
    }
    
    public String getBaseIndexName() {
        return baseIndexName;
    }
    
    public String getAliasName() {
        return aliasName;
    }
    
    public String getActualIndexName() {
        return actualIndexName;
    }
    
    public String getMonthString() {
        return yearMonth.format(MONTH_FORMATTER);
    }
    
    public String getIndexSuffix() {
        return yearMonth.format(INDEX_FORMATTER);
    }
    
    /**
     * 获取当前月份的分区信息
     */
    public static TimePartitionInfo current(String baseIndexName) {
        return new TimePartitionInfo(baseIndexName, YearMonth.now());
    }
    
    /**
     * 获取指定时间的分区信息
     */
    public static TimePartitionInfo of(String baseIndexName, LocalDateTime businessTime) {
        return new TimePartitionInfo(baseIndexName, businessTime);
    }
    
    /**
     * 获取指定年月的分区信息
     */
    public static TimePartitionInfo of(String baseIndexName, int year, int month) {
        return new TimePartitionInfo(baseIndexName, YearMonth.of(year, month));
    }
    
    /**
     * 从索引名称解析分区信息
     */
    public static TimePartitionInfo parse(String indexName) {
        // 假设索引名称格式为：base_name_yyyy_MM
        int lastUnderscoreIndex = indexName.lastIndexOf('_');
        if (lastUnderscoreIndex == -1) {
            throw new IllegalArgumentException("Invalid index name format: " + indexName);
        }
        
        int secondLastUnderscoreIndex = indexName.lastIndexOf('_', lastUnderscoreIndex - 1);
        if (secondLastUnderscoreIndex == -1) {
            throw new IllegalArgumentException("Invalid index name format: " + indexName);
        }
        
        String baseIndexName = indexName.substring(0, secondLastUnderscoreIndex);
        String yearStr = indexName.substring(secondLastUnderscoreIndex + 1, lastUnderscoreIndex);
        String monthStr = indexName.substring(lastUnderscoreIndex + 1);
        
        try {
            int year = Integer.parseInt(yearStr);
            int month = Integer.parseInt(monthStr);
            return new TimePartitionInfo(baseIndexName, YearMonth.of(year, month));
        } catch (NumberFormatException e) {
            throw new IllegalArgumentException("Invalid index name format: " + indexName, e);
        }
    }
    
    /**
     * 获取下一个月的分区信息
     */
    public TimePartitionInfo nextMonth() {
        return new TimePartitionInfo(baseIndexName, yearMonth.plusMonths(1));
    }
    
    /**
     * 获取上一个月的分区信息
     */
    public TimePartitionInfo previousMonth() {
        return new TimePartitionInfo(baseIndexName, yearMonth.minusMonths(1));
    }
    
    /**
     * 检查是否为当前月份
     */
    public boolean isCurrentMonth() {
        return yearMonth.equals(YearMonth.now());
    }
    
    /**
     * 检查是否为指定时间的月份
     */
    public boolean isMonthOf(LocalDateTime dateTime) {
        return yearMonth.equals(YearMonth.from(dateTime));
    }
    
    /**
     * 获取月份的开始时间
     */
    public LocalDateTime getMonthStart() {
        return yearMonth.atDay(1).atStartOfDay();
    }
    
    /**
     * 获取月份的结束时间
     */
    public LocalDateTime getMonthEnd() {
        return yearMonth.atEndOfMonth().atTime(23, 59, 59, 999999999);
    }
    
    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        TimePartitionInfo that = (TimePartitionInfo) o;
        return Objects.equals(yearMonth, that.yearMonth) &&
               Objects.equals(baseIndexName, that.baseIndexName);
    }
    
    @Override
    public int hashCode() {
        return Objects.hash(yearMonth, baseIndexName);
    }
    
    @Override
    public String toString() {
        return "TimePartitionInfo{" +
                "yearMonth=" + yearMonth +
                ", baseIndexName='" + baseIndexName + '\'' +
                ", aliasName='" + aliasName + '\'' +
                ", actualIndexName='" + actualIndexName + '\'' +
                '}';
    }
} 